package cn._51doit.live.udfs;

import cn._51doit.live.pojo.ItemEventCount;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.tuple.Tuple6;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;

public class HotGoodsTopNFunction extends KeyedProcessFunction<Tuple4<String, String, Long, Long>, ItemEventCount, ItemEventCount> {

    private transient ValueState<List<ItemEventCount>> listState;

    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<List<ItemEventCount>> stateDescriptor = new ValueStateDescriptor<>("list-state", TypeInformation.of(new TypeHint<List<ItemEventCount>>() {
        }));
        listState = getRuntimeContext().getState(stateDescriptor);
    }

    @Override
    public void processElement(ItemEventCount value, Context ctx, Collector<ItemEventCount> out) throws Exception {
        //将数据的数据攒起来（为了排序）
        List<ItemEventCount> lst = listState.value();
        if (lst == null) {
            lst = new ArrayList<>();
        }
        lst.add(value);
        listState.update(lst);

        //注册定时器
        ctx.timerService().registerEventTimeTimer(value.windowEnd + 1);

    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<ItemEventCount> out) throws Exception {
        //对数据进行排序
        List<ItemEventCount> lst = listState.value();
        lst.sort((o1, o2) -> Long.compare(o2.count, o1.count));
        for (int i = 0; i < Math.min(lst.size(), 3); i++) {
            ItemEventCount itemEventCount = lst.get(i);
            itemEventCount.order = i + 1;
            out.collect(itemEventCount);
        }
    }
}
